package com.yunchao.library.http

import kotlinx.coroutines.*
import kotlinx.coroutines.channels.BufferOverflow
import kotlinx.coroutines.flow.*
import kotlinx.coroutines.sync.*
import timber.log.Timber
import java.io.*
import java.net.InetSocketAddress
import java.net.Socket
import java.text.SimpleDateFormat
import java.util.Date
import java.util.Locale
import java.util.concurrent.ConcurrentLinkedQueue
import java.util.concurrent.atomic.AtomicBoolean
import java.util.concurrent.atomic.AtomicInteger
import java.util.concurrent.atomic.AtomicLong
import kotlin.math.*

/**
 * TCP连接配置类
 * @param host 服务器主机地址
 * @param port 服务器端口号
 * @param autoReconnect 是否自动重连，默认true
 * @param heartbeatInterval 心跳间隔时间(毫秒)，默认3000
 * @param connectTimeout 连接超时时间(毫秒)，默认5000
 * @param socketTimeout Socket超时时间(毫秒)，默认0(无限)
 * @param maxReconnectAttempts 最大重连尝试次数，默认Int.MAX_VALUE
 * @param maxReconnectDelay 最大重连延迟时间(毫秒)，默认30000
 * @param reconnectBaseDelay 基础重连延迟时间(毫秒)，默认1000
 * @param largeMessageThreshold 大消息阈值(字节)，超过此值会分块传输，默认5MB
 * @param chunkSize 分块大小(字节)，默认1MB
 * @param maxConcurrentSends 最大并发发送数，默认3
 * @param messageBufferCapacity 消息缓冲区容量，默认64
 */
data class TcpConfig(
    val host: String,
    val port: Int,
    val autoReconnect: Boolean = true,
    val heartbeatInterval: Long = 3000L,
    val connectTimeout: Int = 5_000,
    val socketTimeout: Int = 0,
    val maxReconnectAttempts: Int = Int.MAX_VALUE,
    val maxReconnectDelay: Long = 30_000L,
    val reconnectBaseDelay: Long = 1_000L,
    val largeMessageThreshold: Int = 5 * 1024 * 1024, // 5MB
    val chunkSize: Int = 1 * 1024 * 1024, // 1MB
    val maxConcurrentSends: Int = 3,
    val messageBufferCapacity: Int = 64
)

// ================== 连接状态枚举 ==================
sealed class ConnectionState {
    data object Disconnected : ConnectionState()  // 已断开连接
    data object Connecting : ConnectionState()    // 正在连接中
    data class Connected(val socket: Socket) : ConnectionState()  // 已连接
    data class Error(val exception: Throwable) : ConnectionState() // 连接错误
}

// ================== 消息结构 ==================
data class Message(
    val header: MessageHeader,    // 消息头
    val content: MessageContent   // 消息内容
)

/**
 * 消息头结构
 * @param bodyLength 消息体长度
 * @param direction 消息方向(0:接收,1:发送)
 * @param isPacket 是否为数据包
 * @param indicator 指示器标志
 * @param sequence 消息序列号
 * @param transport 传输类型
 * @param isChunked 是否分块传输
 */
data class MessageHeader(
    val bodyLength: Int,
    val direction: Int,
    val isPacket: Int,
    val indicator: Int,
    val sequence: Long,
    val transport: TransportType,
    val isChunked: Boolean
)

/**
 * 消息内容(密封类)
 */
sealed class MessageContent {
    // 单条消息(非分块)
    data class SingleMessage(val data: ByteArray) : MessageContent() {
        override fun equals(other: Any?) = other is SingleMessage && data.contentEquals(other.data)
        override fun hashCode() = data.contentHashCode()
    }

    // 分块消息
    data class ChunkedMessage(
        val chunks: MutableList<ByteArray>,  // 所有分块的列表
        val totalSize: Int                  // 数据总大小（用于校验）
    ) : MessageContent() {
        override fun equals(other: Any?): Boolean {
            if (this === other) return true
            if (javaClass != other?.javaClass) return false
            other as ChunkedMessage
            return totalSize == other.totalSize &&
                    chunks.size == other.chunks.size &&
                    chunks.zip(other.chunks).all { (a, b) -> a.contentEquals(b) }
        }

        override fun hashCode(): Int {
            var result = chunks.hashCode()
            result = 31 * result + totalSize
            return result
        }
    }
}

// ================== 传输类型枚举 ==================
enum class TransportType(val code: Int) {
    HEARTBEAT(0x0000),    // 心跳
    INIT(0x1001),         // 初始化
    MESSAGE(0x1002),      // 普通消息
    CHECK(0x1004),        // 检查
    ACCESS(0x1005),       // 访问
    DISPATCH(0x2002),     // 分发
    EXECUTE(0x2003),      // 执行
    CONFIRM(0x2004),      // 确认
    METADATA(0x2005),     // 元数据
    COMMAND(0x2006),      // 命令
    SUBMIT(0x4002),       // 提交
    AUTH(0x4003),         // 认证
    POWER(0x8001),        // 电源
    EXCHANGE(0x8002);     // 交换

    companion object {
        // 根据code获取对应的传输类型
        fun fromCode(code: Int): TransportType {
            return entries.firstOrNull { it.code == code } ?: HEARTBEAT
        }
    }
}

// ================== 异常体系 ==================
sealed class SocketClientException(message: String?) : Exception(message) {
    class ProtocolError(message: String) : SocketClientException(message)  // 协议错误
    class MaxReconnectAttemptsReached : SocketClientException("达到最大重连次数") // 达到最大重连次数
    class ConnectionFailed : SocketClientException("未连接到服务器")  // 连接失败
    class Disposed : SocketClientException("客户端已释放")  // 客户端已释放
}

/**
 * 高性能TCP Socket客户端实现
 * @param config TCP连接配置
 * @param networkMonitor 网络状态监视器
 * @param socketFactory Socket工厂方法(默认为创建新Socket)
 * @param metrics Socket指标统计(默认为BasicSocketMetrics)
 */
class TcpSocketClient(
    private val config: TcpConfig, // 默认配置
    private val networkMonitor: NetworkMonitor,
    private val socketFactory: () -> Socket = { Socket() },
    private val metrics: SocketMetrics = BasicSocketMetrics().apply { reset() }
) : Closeable, CoroutineScope by CoroutineScope(SupervisorJob() + Dispatchers.IO) {

    // 单线程发送调度器(确保消息顺序)
    @OptIn(ExperimentalCoroutinesApi::class)
    private val sendDispatcher = Dispatchers.IO.limitedParallelism(1)

    // ================== 状态管理 ==================
    private val _connectionState = MutableStateFlow<ConnectionState>(ConnectionState.Disconnected)
    val connectionState: StateFlow<ConnectionState> = _connectionState.asStateFlow()

    // ================== 核心组件 ==================
    private val closed = AtomicBoolean(false)  // 客户端是否已关闭
    private val connectionLock = Mutex()      // 连接状态变更锁
    private val bufferPool = SmartBufferPool(listOf(1024, 2048, 4096, 8192, 16384)) // 智能缓冲池

    private var heartbeatJob: Job? = null     // 心跳任务
    private var reconnectJob: Job? = null     // 重连任务

    private val reconnectAttempts = AtomicInteger(0) // 当前重连尝试次数

    // 消息接收流(使用共享流避免重复订阅)
    private val _incomingMessages = MutableSharedFlow<Message>(
        extraBufferCapacity = config.messageBufferCapacity,
        onBufferOverflow = BufferOverflow.SUSPEND
    )
    val incomingMessages: SharedFlow<Message> = _incomingMessages.asSharedFlow()

    init {
        // 监听网络状态变化
        launch {
            networkMonitor.isOnline.collect { isOnline ->
                handleConfigAndNetworkChange(isOnline)
            }
        }
    }

    // ================== 网络状态变化处理 ==================
    private suspend fun handleConfigAndNetworkChange(isOnline: Boolean) {
        connectionLock.withLock {
            when {
                !isOnline -> disconnect() // 网络断开时主动断开连接
                isOnline && !isConnected() && config.autoReconnect -> connect() // 网络恢复时自动重连
            }
        }
    }

    // ================== 公开API ==================

    /**
     * 启动客户端
     */
    fun start() {
        checkNotDisposed()
        launch { connect() }
    }

    /**
     * 发送消息
     * @param direction 消息方向
     * @param transport 传输类型
     * @param bodyBytes 消息体字节数组
     * @param isPacket 是否为数据包
     * @param indicator 指示器标志
     * @param sequence 消息序列号
     */
    suspend fun sendMessage(
        direction: Int,
        transport: TransportType,
        bodyBytes: ByteArray,
        isPacket: Int = 0,
        indicator: Int = 0,
        sequence: Long = 0
    ) {
        checkConnected()
        withContext(sendDispatcher) {
            try {
                internalSend(direction, transport, bodyBytes, isPacket, indicator, sequence)
            } catch (e: Exception) {
                handleConnectionError(e)
            }
        }
    }

    /**
     * 关闭客户端
     */
    override fun close() {
        if (closed.compareAndSet(false, true)) {
            launch {
                disconnectInternal()
                cleanupResources()
                coroutineContext.cancel(CancellationException("TcpSocketClient closed"))
            }
            metrics.recordConnectionEvent(ConnectionEvent.Disconnected)
        }
    }

    // ================== 连接管理 ==================

    /**
     * 连接服务器
     */
    private suspend fun connect() = connectionLock.withLock {
        if (shouldSkipConnect()) return@withLock

        updateState(ConnectionState.Connecting)
        metrics.recordConnectionEvent(ConnectionEvent.Connecting)

        try {
            disconnectInternal() // 先断开现有连接
            val socket = socketFactory().apply {
                connect(InetSocketAddress(config.host, config.port), config.connectTimeout)
                keepAlive = true    // 启用TCP keepalive
                tcpNoDelay = true   // 禁用Nagle算法
                soTimeout = 0       // 设置读取超时(0表示无限)
            }
            handleSuccessfulConnection(socket)
        } catch (e: Exception) {
            handleConnectionError(e)
        }
    }

    /**
     * 断开连接
     */
    private suspend fun disconnect() = connectionLock.withLock {
        if (shouldSkipDisconnect()) return@withLock
        disconnectInternal()
    }

    /**
     * 处理成功连接
     */
    private fun handleSuccessfulConnection(socket: Socket) {
        if (closed.get()) return // 防止在连接过程中被关闭
        closed.set(false)

        startHeartbeat() // 启动心跳
        updateState(ConnectionState.Connected(socket))
        reconnectAttempts.set(0) // 重置重连计数器
        metrics.recordConnectionEvent(ConnectionEvent.Connected(config.host, config.port))
        receiveMessages(socket) // 开始接收消息
    }

    // ================== 消息处理 ==================

    /**
     * 内部发送消息实现
     */
    private fun internalSend(
        direction: Int,
        transport: TransportType,
        bodyBytes: ByteArray,
        isPacket: Int,
        indicator: Int,
        sequence: Long
    ) {
        val isChunked = bodyBytes.size > config.largeMessageThreshold
        val header = MessageHeader(
            bodyLength = bodyBytes.size,
            direction = direction,
            isPacket = isPacket,
            indicator = indicator,
            sequence = sequence,
            transport = transport,
            isChunked = isChunked
        )
        metrics.recordMessageSent(header, bodyBytes)

        val socket = (connectionState.value as ConnectionState.Connected).socket
        val outputStream = socket.getOutputStream()

        outputStream.write(buildHeaderBytes(header)) // 写入消息头

        // 根据消息大小决定是否分块发送
        when {
            isChunked -> sendInChunks(outputStream, bodyBytes)
            else -> outputStream.write(bodyBytes)
        }
        outputStream.flush()
    }


    /**
     * 接收消息
     */
    private fun receiveMessages(socket: Socket) = launch {
        val inputStream = socket.getInputStream()
        val headerBuffer = ByteArray(14) // 消息头固定14字节

        while (isActive && isConnected()) {
            try {
                readFully(inputStream, headerBuffer)
                val header = parseHeader(headerBuffer)

                val message = if (header.isChunked) {
                    receiveChunkedMessage(inputStream, header) // 接收分块消息
                } else {
                    Message(
                        header,
                        MessageContent.SingleMessage(
                            readBody(
                                inputStream,
                                header.bodyLength
                            )
                        ) // 接收普通消息
                    )
                }
                _incomingMessages.emit(message).also {
                    metrics.recordMessageReceived(message)
                }
            } catch (e: Exception) {
                if (isConnected()) handleConnectionError(e)
            }
        }
    }

    // ================== 协议处理 ==================

    /**
     * 解析消息头
     */
    private fun parseHeader(headerBytes: ByteArray): MessageHeader {
        if (headerBytes.size != 14) {
            throw SocketClientException.ProtocolError("无效的消息头长度")
        }
        DataInputStream(ByteArrayInputStream(headerBytes)).use { dis ->
            val firstFourBytes = dis.readInt()
            val direction = (firstFourBytes shr 31) and 1
            val isPacket = (firstFourBytes shr 30) and 1
            val indicator = (firstFourBytes shr 28) and 0x3
            val bodyLength = firstFourBytes and 0x0FFFFFFF
            val sequence = dis.readLong()
            val byte1 = dis.readByte().toInt() and 0xFF
            val byte2 = dis.readByte().toInt() and 0xFF
            val transportCode = (byte1 shl 8) or byte2
            val transport = TransportType.fromCode(transportCode)
            val isChunked = bodyLength > config.largeMessageThreshold
            return MessageHeader(
                bodyLength = bodyLength,
                direction = direction,
                isPacket = isPacket,
                indicator = indicator,
                sequence = sequence,
                transport = transport,
                isChunked = isChunked
            )
        }
    }

    /**
     * 构建消息头字节数组
     */
    private fun buildHeaderBytes(header: MessageHeader): ByteArray {
        var firstFourBytes = header.bodyLength
        firstFourBytes = firstFourBytes or (header.direction shl 31)
        firstFourBytes = firstFourBytes or (header.isPacket shl 30)
        firstFourBytes = firstFourBytes or (header.indicator shl 28)

        ByteArrayOutputStream(14).use { baos ->
            DataOutputStream(baos).use { dos ->
                dos.writeInt(firstFourBytes)
                dos.writeLong(header.sequence)
                val transportCode = header.transport.code
                dos.writeByte(transportCode ushr 8)
                dos.writeByte(transportCode and 0xFF)
                return baos.toByteArray()
            }
        }
    }

    // ================== 分块传输 ==================

    /**
     * 分块发送消息
     */
    private fun sendInChunks(outputStream: OutputStream, data: ByteArray) {
        val buffer = bufferPool.acquire(config.chunkSize)
        try {
            var offset = 0
            while (offset < data.size && isConnected()) {
                val chunkSize = min(buffer.size, data.size - offset)
                System.arraycopy(data, offset, buffer, 0, chunkSize)
                outputStream.write(buffer, 0, chunkSize)

                metrics.recordChunkSent(offset, chunkSize, data)
                offset += chunkSize
            }
        } finally {
            bufferPool.release(buffer)
        }
    }

    /**
     * 接收分块消息
     */
    private fun receiveChunkedMessage(inputStream: InputStream, header: MessageHeader): Message {
        val chunks = mutableListOf<ByteArray>()
        var remaining = header.bodyLength
        val buffer = bufferPool.acquire(config.chunkSize)

        try {
            while (remaining > 0) {
                val readSize = min(buffer.size, remaining)
                readFully(inputStream, buffer, readSize)
                chunks.add(buffer.copyOf(readSize))
                remaining -= readSize
            }

            return Message(
                header = header,
                content = MessageContent.ChunkedMessage(
                    chunks = chunks,
                    totalSize = header.bodyLength
                )
            )
        } finally {
            bufferPool.release(buffer)
        }
    }

    // ================== 工具方法 ==================

    private fun updateState(newState: ConnectionState) {
        _connectionState.value = newState
    }

    private fun checkConnected() {
        if (closed.get() || !isConnected()) {
            throw SocketClientException.ConnectionFailed()
        }
    }

    private fun checkNotDisposed() {
        if (closed.get()) throw SocketClientException.Disposed()
    }

    private fun isConnected() = connectionState.value is ConnectionState.Connected

    private fun shouldSkipConnect(): Boolean {
        return closed.get() || connectionState.value.let {
            it is ConnectionState.Connected || it is ConnectionState.Connecting
        }
    }

    private fun shouldSkipDisconnect(): Boolean {
        return closed.get() || connectionState.value !is ConnectionState.Connected
    }

    /**
     * 从输入流中完整读取指定长度的数据
     */
    private fun readFully(input: InputStream, buffer: ByteArray, length: Int = buffer.size) {
        var offset = 0
        while (offset < length) {
            val read = input.read(buffer, offset, length - offset)
            if (read == -1) throw EOFException("意外的流结束")
            offset += read
        }
    }

    /**
     * 读取消息体
     */
    private fun readBody(input: InputStream, length: Int): ByteArray {
        return ByteArray(length).also {
            readFully(input, it)
        }
    }

    // ================== 心跳机制 ==================

    /**
     * 启动心跳
     */
    private fun startHeartbeat() {
        heartbeatJob?.cancel()
        heartbeatJob = launch {
            while (isActive && isConnected()) {
                delay(config.heartbeatInterval)
                try {
                    internalSend(
                        direction = 1,
                        transport = TransportType.HEARTBEAT,
                        bodyBytes = byteArrayOf(),
                        isPacket = 0,
                        indicator = 0,
                        sequence = 0
                    )
                } catch (e: Exception) {
                    handleConnectionError(e)
                }
            }
        }
    }

    // ================== 错误处理 ==================

    /**
     * 处理连接错误
     */
    private fun handleConnectionError(error: Throwable) {
        if (closed.get()) return

        launch {
            connectionLock.withLock {
                disconnectInternal()
                when {
                    !config.autoReconnect -> {
                        updateState(ConnectionState.Error(error))
                        metrics.recordConnectionEvent(ConnectionEvent.Error(error))
                    }

                    reconnectAttempts.get() >= config.maxReconnectAttempts -> {
                        updateState(ConnectionState.Error(SocketClientException.MaxReconnectAttemptsReached()))
                        metrics.recordConnectionEvent(ConnectionEvent.Error(SocketClientException.MaxReconnectAttemptsReached()))
                    }

                    else -> scheduleReconnect()
                }
            }
        }
    }

    /**
     * 安排重连
     */
    private fun scheduleReconnect() {
        reconnectAttempts.incrementAndGet()
        // 使用指数退避算法计算重连延迟
        val delayMs = min(
            config.reconnectBaseDelay * 2.0.pow(reconnectAttempts.get().toDouble() - 1).toLong(),
            config.maxReconnectDelay
        )
        metrics.recordConnectionEvent(
            ConnectionEvent.Reconnecting(reconnectAttempts.get(), delayMs)
        )
        reconnectJob?.cancel()

        reconnectJob = launch {
            SocketLogger.logOperation("等待重连", "延迟: ${delayMs}ms")
            delay(delayMs)
            connect()
        }
    }

    // ================== 资源管理 ==================

    /**
     * 清理资源
     */
    private fun cleanupResources() {
        disconnectInternal()
        bufferPool.clear()
    }

    /**
     * 内部断开连接实现
     */
    private fun disconnectInternal() {
        (connectionState.value as? ConnectionState.Connected)?.socket?.close()
        updateState(ConnectionState.Disconnected)
        heartbeatJob?.cancel()
        reconnectJob?.cancel()
    }

    // ================== 内部类 ==================

    /**
     * 智能缓冲池(减少内存分配开销)
     */
    private class SmartBufferPool(private val sizes: List<Int>) {
        private val pools = sizes.associateWith { ConcurrentLinkedQueue<ByteArray>() }

        // 获取缓冲区
        fun acquire(size: Int): ByteArray {
            val suitableSize = sizes.firstOrNull { it >= size } ?: size
            return pools[suitableSize]?.poll() ?: ByteArray(suitableSize)
        }

        // 释放缓冲区
        fun release(buffer: ByteArray) {
            pools[buffer.size]?.offer(buffer)
        }

        // 清空缓冲池
        fun clear() {
            pools.values.forEach { it.clear() }
        }
    }
}

// ================== 接口定义 ==================

/**
 * Socket指标统计接口
 */
interface SocketMetrics {
    fun recordMessageSent(header: MessageHeader, bodyBytes: ByteArray) // 记录消息发送
    fun recordMessageReceived(message: Message) // 记录消息接收
    fun recordChunkSent(offset: Int, chunkSize: Int, bodyBytes: ByteArray) // 记录分块发送
    fun recordConnectionEvent(event: ConnectionEvent) // 记录连接事件
}

/**
 * 连接事件枚举
 */
sealed class ConnectionEvent {
    data object Connecting : ConnectionEvent() // 正在连接
    data class Connected(val host: String, val port: Int) : ConnectionEvent() // 连接成功
    data object Disconnected : ConnectionEvent() // 连接断开
    data class Reconnecting(val attempt: Int, val delayMs: Long) : ConnectionEvent() // 正在重连
    data class Error(val cause: Throwable) : ConnectionEvent() // 连接错误
}

/**
 * 基础Socket指标统计实现
 */
class BasicSocketMetrics : SocketMetrics {
    private val messagesSent = AtomicInteger(0) // 发送消息数
    private val messagesReceived = AtomicInteger(0) // 接收消息数
    private val bytesSent = AtomicLong(0) // 发送字节数
    private val bytesReceived = AtomicLong(0) // 接收字节数
    private val chunksSent = AtomicInteger(0) // 发送分块数
    private val chunksReceived = AtomicInteger(0) // 接收分块数
    private val chunkedBytesSent = AtomicLong(0) // 分块发送字节数
    private val chunkedBytesReceived = AtomicLong(0) // 分块接收字节数
    private val connectionAttempts = AtomicInteger(0) // 连接尝试次数
    private val connectionSuccesses = AtomicInteger(0) // 连接成功次数
    private val connectionFailures = AtomicInteger(0) // 连接失败次数
    private val errorCounts = mutableMapOf<String, AtomicInteger>() // 错误统计

    override fun recordMessageSent(header: MessageHeader, bodyBytes: ByteArray) {
        messagesSent.incrementAndGet()
        bytesSent.addAndGet(header.bodyLength.toLong())
        SocketLogger.logMessage("SEND", header, bodyBytes)
    }

    override fun recordMessageReceived(message: Message) {
        if (message.header.isChunked) {
            chunksReceived.incrementAndGet()
            chunkedBytesReceived.addAndGet(message.header.bodyLength.toLong())
        } else {
            messagesReceived.incrementAndGet()
            bytesReceived.addAndGet(message.header.bodyLength.toLong())
        }
        SocketLogger.logMessage("RECEIVE", message)
    }

    override fun recordChunkSent(offset: Int, chunkSize: Int, bodyBytes: ByteArray) {
        chunksSent.incrementAndGet()
        chunkedBytesSent.addAndGet(bodyBytes.size.toLong())
        SocketLogger.logOperation(
            "发送分块 offset = $offset 分块大小 = $chunkSize 总计 = ${bodyBytes.size} 字节"
        )
    }

    override fun recordConnectionEvent(event: ConnectionEvent) {
        SocketLogger.logConnectionState(event)
        when (event) {
            is ConnectionEvent.Connecting -> connectionAttempts.incrementAndGet()
            is ConnectionEvent.Connected -> connectionSuccesses.incrementAndGet()
            is ConnectionEvent.Disconnected -> connectionFailures.incrementAndGet()
            is ConnectionEvent.Reconnecting -> connectionAttempts.incrementAndGet()
            is ConnectionEvent.Error -> connectionFailures.incrementAndGet()
        }
    }

    /**
     * 记录错误
     */
    fun recordError(errorType: String) {
        errorCounts.computeIfAbsent(errorType) { AtomicInteger(0) }.incrementAndGet()
    }

    /**
     * 获取统计信息
     */
    fun getStats(): SocketStats {
        return SocketStats(
            messagesSent = messagesSent.get(),
            messagesReceived = messagesReceived.get(),
            bytesSent = bytesSent.get(),
            bytesReceived = bytesReceived.get(),
            chunksSent = chunksSent.get(),
            chunksReceived = chunksReceived.get(),
            chunkedBytesSent = chunkedBytesSent.get(),
            chunkedBytesReceived = chunkedBytesReceived.get(),
            connectionAttempts = connectionAttempts.get(),
            connectionSuccesses = connectionSuccesses.get(),
            connectionFailures = connectionFailures.get(),
            errorCounts = errorCounts.mapValues { it.value.get() },
            timestamp = System.currentTimeMillis()
        )
    }

    /**
     * 重置统计信息
     */
    fun reset() {
        messagesSent.set(0)
        messagesReceived.set(0)
        bytesSent.set(0)
        bytesReceived.set(0)
        chunksSent.set(0)
        chunksReceived.set(0)
        chunkedBytesSent.set(0)
        chunkedBytesReceived.set(0)
        connectionAttempts.set(0)
        connectionSuccesses.set(0)
        connectionFailures.set(0)
        errorCounts.clear()
    }

    /**
     * Socket统计信息数据类
     */
    data class SocketStats(
        val messagesSent: Int,
        val messagesReceived: Int,
        val bytesSent: Long,
        val bytesReceived: Long,
        val chunksSent: Int,
        val chunksReceived: Int,
        val chunkedBytesSent: Long,
        val chunkedBytesReceived: Long,
        val connectionAttempts: Int,
        val connectionSuccesses: Int,
        val connectionFailures: Int,
        val errorCounts: Map<String, Int>,
        val timestamp: Long
    ) {
        // 计算连接成功率
        private fun connectionSuccessRate(): Double {
            val total = connectionAttempts
            return if (total > 0) connectionSuccesses.toDouble() / total else 0.0
        }

        // 计算平均发送消息大小
        private fun averageSentMessageSize(): Int {
            return if (messagesSent > 0) (bytesSent.toDouble() / messagesSent).roundToInt() else 0
        }

        // 计算分块传输比例
        private fun chunkedTransferRatio(): Double {
            val totalBytes = bytesSent + bytesReceived
            val chunkedBytes = chunkedBytesSent + chunkedBytesReceived
            return if (totalBytes > 0) chunkedBytes.toDouble() / totalBytes else 0.0
        }

        override fun toString(): String {
            return """
            |==== Socket 统计报告 ====
            |时间: ${Date(timestamp)}
            |
            |== 消息统计 ==
            |发送消息数: $messagesSent (${bytesSent.toHumanReadableSize()})
            |接收消息数: $messagesReceived (${bytesReceived.toHumanReadableSize()})
            |平均发送大小: ${averageSentMessageSize().toHumanReadableSize()}
            |
            |== 分块传输 ==
            |分块发送: $chunksSent (${chunkedBytesSent.toHumanReadableSize()})
            |分块接收: $chunksReceived (${chunkedBytesReceived.toHumanReadableSize()})
            |分块比例: ${"%.1f".format(chunkedTransferRatio() * 100)}%
            |
            |== 连接统计 ==
            |连接尝试: $connectionAttempts
            |成功连接: $connectionSuccesses
            |连接失败: $connectionFailures
            |成功率: ${"%.1f".format(connectionSuccessRate() * 100)}%
            |
            |== 错误统计 ==
            |${errorCounts.entries.joinToString("\n") { "  ${it.key}: ${it.value}" }}
            |=========================
            """.trimMargin()
        }
    }
}

// ================== 扩展函数 ==================

/**
 * 将字节数转换为可读格式
 */
private fun Long.toHumanReadableSize(): String {
    if (this <= 0) return "0 B"
    val units = arrayOf("B", "KB", "MB", "GB", "TB")
    val digitGroups = (log10(toDouble()) / log10(1024.0)).toInt()
    return "%.1f %s".format(this / 1024.0.pow(digitGroups.toDouble()), units[digitGroups])
}

private fun Int.toHumanReadableSize(): String = toLong().toHumanReadableSize()

// ================== Socket日志工具 ==================

/**
 * Socket日志工具类
 */
object SocketLogger {
    private const val TAG = "SocketTracing"

    // 使用线程安全的日期格式化
    private val dateFormat by lazy {
        ThreadLocal.withInitial {
            SimpleDateFormat("HH:mm:ss.SSS", Locale.getDefault())
        }.get()
    }

    /**
     * 记录消息日志
     */
    fun logMessage(direction: String, message: Message) {
        val timestamp = dateFormat?.format(System.currentTimeMillis())
        val threadName = Thread.currentThread().name

        Timber.tag(TAG).d(
            """
              [$timestamp][$threadName] $direction 消息:
                    消息头: ${message.header} 
                    传输类型: ${message.header.transport.code.toString(16)}
                    负载: ${
                when (val content = message.content) {
                    is MessageContent.ChunkedMessage -> {
                        ("分块(${content.chunks.size}/${content.totalSize}) ")
                    }

                    is MessageContent.SingleMessage -> {
                        ("十六进制=${content.data} ")
                    }
                }
            }
            """.trimIndent()
        )
    }

    /**
     * 记录消息日志(重载)
     */
    fun logMessage(
        direction: String,
        header: MessageHeader,
        bodyBytes: ByteArray,
    ) {
        val timestamp = dateFormat?.format(System.currentTimeMillis())
        val threadName = Thread.currentThread().name

        Timber.tag(TAG).v(
            """
              [$timestamp][$threadName] $direction 消息:
                   消息头: $header 
                   传输类型: ${header.transport.code.toString(16)}
                   负载: $bodyBytes
            """.trimIndent()
        )
    }

    /**
     * 记录连接状态
     */
    fun logConnectionState(event: ConnectionEvent) {
        val timestamp = dateFormat?.format(System.currentTimeMillis())
        val threadName = Thread.currentThread().name

        val message = when (event) {
            is ConnectionEvent.Connecting -> "正在连接服务器..."
            is ConnectionEvent.Connected -> "成功连接到 ${event.host}:${event.port}"
            is ConnectionEvent.Disconnected -> "与服务器断开连接"
            is ConnectionEvent.Reconnecting -> "正在尝试第${event.attempt}次重新连接，延迟${event.delayMs}毫秒"
            is ConnectionEvent.Error -> "连接发生错误 ${event.cause.message}"
        }

        Timber.tag(TAG).i("[$timestamp][$threadName] 连接状态: $message")
    }

    /**
     * 记录操作日志
     */
    fun logOperation(operation: String, details: String? = null) {
        val timestamp = dateFormat?.format(System.currentTimeMillis())
        val threadName = Thread.currentThread().name

        val message = buildString {
            append("[$timestamp][$threadName] 操作: $operation")
            details?.let { append(" - $it") }
        }

        Timber.tag(TAG).d(message)
    }

    /**
     * 记录错误日志
     */
    fun logError(errorMsg: String, exception: Throwable? = null) {
        val timestamp = dateFormat?.format(System.currentTimeMillis())
        val threadName = Thread.currentThread().name
        val message = "[$timestamp][$threadName] 错误: $errorMsg"

        if (exception != null) {
            Timber.tag(TAG).e(exception, message)
        } else {
            Timber.tag(TAG).e(message)
        }
    }

    /**
     * 记录警告日志
     */
    fun logWarning(warningMsg: String, exception: Throwable? = null) {
        val timestamp = dateFormat?.format(System.currentTimeMillis())
        val threadName = Thread.currentThread().name
        val message = "[$timestamp][$threadName] 警告: $warningMsg"

        if (exception != null) {
            Timber.tag(TAG).w(exception, message)
        } else {
            Timber.tag(TAG).w(message)
        }
    }
}

